/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.core.redis;

import io.netty.channel.nio.NioEventLoopGroup;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.semidata.rtd.core.service.Service;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class RedisClient implements Service {

	private static final long CHECK_TIME = 2000;
	private RedisConnectionPool[] connectionPools;
	private NioEventLoopGroup group;
	private Thread thread = null;
	private boolean quit = false;
	private ReentrantLock lock = new ReentrantLock();
	private Condition cond = lock.newCondition();
	private Thread checkThread = null;

	public RedisClient(String[] addrs, int threadNum, int connectionNum,
	        long timeout) {

		connectionPools = new RedisConnectionPool[addrs.length];
		group = new NioEventLoopGroup(threadNum);

		for (int i = 0; i < addrs.length; i++) {

			int pos = addrs[i].indexOf(':');
			if (pos == -1)
				throw new RuntimeException("invalid redis addr");

			String hostname = addrs[i].substring(0, pos);
			int port;
			try {
				port = Integer.parseInt(addrs[i].substring(pos + 1));
			} catch (Exception e) {
				throw new RuntimeException("invalid redis addr");
			}

			connectionPools[i] = new RedisConnectionPool(
			        "RedisConnection-" + i, group, hostname, port, this,
			        connectionNum, timeout);
		}
	}

	public void signal() {
		lock.lock();
		try {
			cond.signal();
		} finally {
			lock.unlock();
		}
	}

	private void schedule() {

		while (!quit) {

			lock.lock();
			try {

				try {
					cond.await(5, TimeUnit.MILLISECONDS);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			} finally {
				lock.unlock();
			}

			for (RedisConnectionPool pool : connectionPools) {
				pool.send();
			}

		}
	}

	private void check() {

		while (!quit) {

			for (RedisConnectionPool pool : connectionPools) {
				pool.check();
			}
			try {
				Thread.sleep(CHECK_TIME);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public void start() {
		
		for (RedisConnectionPool pool : connectionPools) {
			pool.start();
		}

		thread = new Thread(new Runnable() {

			@Override
			public void run() {
				schedule();
			}
		});

		thread.start();

		checkThread = new Thread(new Runnable() {

			@Override
            public void run() {
				check();
            }
			
		});
		checkThread.start();
	}

	@Override
	public void stop() {
		quit = true;
		signal();
		if (thread != null) {
			try {
				thread.join();
			} catch (InterruptedException e) {
			}
		}
		
		if (checkThread != null) {
			try {
				checkThread.join();
			} catch (InterruptedException e) {
			}
		}

		for (RedisConnectionPool pool : connectionPools) {
			pool.stop();
		}

		group.shutdownGracefully();
	}

	public void get(String key, String field, long timeout,
	        RedisResponseHandler handler) {
		if (key == null || field == null || handler == null)
			return;

		RedisConnectionPool pool = getConnectionPool(key);
		if (pool == null) {
			handler.handle(null, new NullPointerException());
		} else {
			pool.get(key.getBytes(), field.getBytes(), timeout, handler);
		}
	}
	
	public void getAll(String key, long timeout,
	        RedisResponseHandler handler) {
		if (key == null  || handler == null)
			return;

		RedisConnectionPool pool = getConnectionPool(key);
		if (pool == null) {
			handler.handle(null, new NullPointerException());
		} else {
			pool.getAll(key.getBytes(), timeout, handler);
		}
	}

	public void set(String key, String field, String value, long timeout,
	        RedisResponseHandler handler) {
		if (key == null || field == null || handler == null)
			return;

		RedisConnectionPool pool = getConnectionPool(key);
		if (pool == null) {
			handler.handle(null, new NullPointerException());
		} else {
			pool.set(key.getBytes(), field.getBytes(), value.getBytes(),
			        timeout, handler);
		}

	}

	public void expire(String key, long time, long timeout,
	        RedisResponseHandler handler) {
		if (key == null || handler == null)
			return;

		RedisConnectionPool pool = getConnectionPool(key);
		if (pool == null) {
			handler.handle(null, new NullPointerException());
		} else {
			pool.expire(key.getBytes(), time, timeout, handler);
		}
	}
	
	public void del(String key, long timeout,
	        RedisResponseHandler handler) {
		if (key == null || handler == null)
			return;

		RedisConnectionPool pool = getConnectionPool(key);
		if (pool == null) {
			handler.handle(null, new NullPointerException());
		} else {
			pool.del(key.getBytes(), timeout, handler);
		}
	}

	public RedisConnectionPool getConnectionPool(String key) {

		if (key == null || connectionPools == null
		        || connectionPools.length == 0)
			return null;

		int index = key.hashCode() % connectionPools.length;
		if (index < 0)
			index = connectionPools.length + index;
		
		return connectionPools[index];

	}
}
